package main

import (
	"DMSKafka/config"
	"context"
	"github.com/Shopify/sarama"
	"github.com/pkg/errors"
	"log"
)

// 真实Topic名称
var realTopic = "realque"

// 消费者组
func main() {
	conf, err := config.GetConfig()
	if err != nil {
		log.Printf("%+v\n", err)
		return
	}
	//初始化消费者组
	config := sarama.NewConfig()
	config.Consumer.Return.Errors = false
	//创建消费者组
	group, err := sarama.NewConsumerGroup(conf.Host, "real_group", config)
	if err != nil {
		log.Printf("%+v\n", errors.Wrap(err, "创建消费者组失败"))
		return
	}
	defer group.Close()
	go func() {
		for err := range group.Errors() {
			log.Printf("%+v\n", err)
		}
	}()
	ctx := context.Background()
	for {
		topic := []string{realTopic}
		handler := Consumer{}
		err := group.Consume(ctx, topic, handler)
		if err != nil {
			log.Printf("%+v\n", err)
			return
		}
	}
}

// 实现ConsumerGroupHandler接口
type Consumer struct {
}

func (Consumer) Setup(sarama.ConsumerGroupSession) error {
	return nil
}

func (Consumer) Cleanup(sarama.ConsumerGroupSession) error {
	return nil
}

func (Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	for message := range claim.Messages() {
		log.Printf("最终的Message: value = %s, timestamp = %v, topic = %s", string(message.Value), message.Timestamp, message.Topic)
		session.MarkMessage(message, "")
	}
	return nil
}
